package cn.hhz.charging.order.mqtt;

import cn.hhz.charging.common.pojo.JsonResult;
import cn.hhz.charging.common.pojo.param.ChargingProcessParam;
import cn.hhz.charging.common.pojo.vo.ChargingProcessVO;
import cn.hhz.charging.common.utils.JsonUtils;
import cn.hhz.charging.order.constant.Constant;
import cn.hhz.charging.order.constant.MqttConstant;
import cn.hhz.charging.order.constant.OrderStatusConstant;
import cn.hhz.charging.order.pojo.dto.ChargingProcessDto;
import cn.hhz.charging.order.pojo.dto.ChargingResultDto;
import cn.hhz.charging.order.pojo.po.ChargingBillFailPO;
import cn.hhz.charging.order.pojo.po.ChargingBillSuccessPO;
import cn.hhz.charging.order.pojo.po.ChargingProcessPO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;

import java.math.BigDecimal;

/**
 * 用来消费设备发送的消息
 */
@Slf4j
public class MqttConsumer implements MqttCallbackExtended {

    private MqttClient mqttClient;
    private MqttContext mqttContext;

    public MqttConsumer(MqttContext mqttContext, MqttClient mqttClient) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    /**
     * 连接完成,表示订单服务和EMQX消息中间件连接成功
     *
     * @param reconnect If true, the connection was the result of automatic reconnect.
     * @param serverURI The server URI that the connection was made to.
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.debug("MqttConsumer#connectComplete方法被调用了,表示连接成功");
        //连接成功 订阅 topic 来接收设备给订单发送的消息
        try {
            log.debug("连接成功,订阅topic:{},用来接收设备发送的充电结果信息", MqttConstant.TOPIC_CHARGING_RESULT);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_RESULT);
            log.debug("连接成功,订阅topic:{} 成功,用来接收设备发送的充电结果信息", MqttConstant.TOPIC_CHARGING_RESULT);
            log.debug("连接成功,订阅topic:{},用来接收设备同步的充电状态信息", MqttConstant.TOPIC_CHARGING_PROCESS);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_PROCESS);
            log.debug("连接成功,订阅topic:{} 成功,用来接收设备同步的充电状态信息", MqttConstant.TOPIC_CHARGING_PROCESS);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 连接丢失
     *
     * @param cause the reason behind the loss of connection.
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.debug("MqttConsumer#connectionLost方法被调用了,表示连接丢失");
    }

    /**
     * 消息到达,不是订单服务的消息到达设备
     * 设备发送消息,到达了订单服务
     *
     * @param topic   name of the topic on the message was published to
     * @param message the actual message.
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("MqttConsumer#messageArrived方法被调用了,消息到达,topic:{},message:{}", topic, message);
        if (MqttConstant.TOPIC_CHARGING_RESULT.equals(topic)) {
            log.debug("处理开始充电结果消息");
            handleChargingResult(message);
        } else if (MqttConstant.TOPIC_CHARGING_PROCESS.equals(topic)) {
            log.debug("处理开始充电进度消息");
            handleChargingProcess(message);
        }

    }

    /**
     * 处理开始充电进度消息
     *
     * @param message
     */
    private void handleChargingProcess(MqttMessage message) {
        log.debug("处理充电进度:{}",message);
        ChargingProcessParam chargingProcessParam = null;
        try{
            //把设备发送的消息充电进度JSON转换java对象ChargingProcessDto
            chargingProcessParam =
                    JsonUtils.fromJson(message.toString(),ChargingProcessParam.class);
            log.debug("充电进度消息转换为java对象:{}",chargingProcessParam);
        }catch (Exception e){
            log.error("设备发送的充电进度消息格式{}有问题,请检查!",message,e);
        }
        //1 调用计价服务计算价格
        log.debug("通过open-feign的costClient 调用计价服务,计算价格,入参:{}",chargingProcessParam);
        ChargingProcessVO orderCost = getOrderCost(chargingProcessParam);
        log.debug("通过open-feign的costClient 调用计价服务,计算价格,入参:{},出参:{}",chargingProcessParam,orderCost);
        if (orderCost != null) {
            //2 充电进度保存
            saveChargingProcess(chargingProcessParam);
            //3 推送充电数据给用户 推送 websocket
            syncChargingProcess2User(orderCost);
            //4 充电过程检查用户余额
            // 钱够 继续充
            // 钱不够
            //   4.1 给用户发通知,钱不够,引导用户充值  推荐
            //   4.2 钱不够 继续充,直到充满为止,记录下用户欠款,等用户下次再充的时候,需要结清欠款
            BigDecimal totalCost = orderCost.getTotalCost();
            log.debug("当前花费金额:{}",totalCost);
            checkUserBalance(chargingProcessParam.getUserId(),totalCost);
            //5 判断温度
            //  5.1 大于 40 度 小于 50度 ,B级告警, 推送给用户,用户可以选择是否继续充
            //  5.2 大于 50  A级告警, 推送给用户,停止充电
            checkDeviceStatus(chargingProcessParam.getTemperature());
            //6 判断是否充满
            // 6.1 设备停止充电,更新订单状态 为 订单结束,扣除用户余额
            checkChargingStatus(chargingProcessParam.getIsFull());
        }else {
            log.error("获取价格失败:{},打印错误日志,告警",chargingProcessParam);
        }
    }

    /**
     * 检查是否充满
     * @param isFull
     */
    private void checkChargingStatus(Boolean isFull) {

    }

    /**
     * 检查设备温度
     * @param temperature
     */
    private void checkDeviceStatus(float temperature) {

    }

    /**
     * 检查用户余额
     * @param userId
     * @param totalCost
     */
    private void checkUserBalance(Integer userId, BigDecimal totalCost) {

    }

    /**
     * 通过充电进度数据给用户
     * @param orderCost
     */
    private void syncChargingProcess2User(ChargingProcessVO orderCost) {
        log.debug("同步充电数据到用户手机端:{}",orderCost);
        String json = JsonUtils.toJson(orderCost);
        mqttContext.getSocketServer().sendMessage(orderCost.getUserId(),json);
        //把消息发送给RabbitMQ,
        // 专门搞一个
        // 消息队列存储消息(充电进度),
        // 1 混合存 用户不多 把告警信息,充电进度信息等等用一个队列
        // [,5,4,3-告警,2-告警,1-同]  相互阻塞
        // 2 分开存 用户多 把告警信息一个队列,充电进度一个队列,直接不会相互影响
        // [告警,2-告警,]
        // [,5-同,1-同]
        // 再搞一个消费者消费消息
        //消费者怎么消费? 业务不关心,调用webSocket,集成服务商(极光推送,网易云信) 给用户发送信息
    }

    /**
     * 保存充电记录
     * 数据量很大
     * 设备定时同步 30秒同步一次 1分同步两次
     * 1天有1000个用户充电 1000个订单
     * 1个订单充平均4个小时电  有慢充,有快充
     * 1天会产生多少数据 ?
     *
     * 1天 48W数据
     * 1年 48W * 365
     * 10年 48W * 365 * 10
     *
     * 1 都把数据存储到表里
     * 2 冷热数据分离 charging_bill_success
     * 冷 就是不经常使用的数据 归档
     * 热 经常使用  1个月-半年  1个月-3个月 1个月-半年 1个月-一年
     *  charging_bill_success 表里 只存储 半年的数据 热数据  机器配置高
     *
     *  charging_bill_success_08
     *  charging_bill_success_09
     *  charging_bill_success_10
     *  charging_bill_success_11
     *  charging_bill_success_12
     *
     *  按照月份存储
     *  用户的app 我的订单 订单列表
     *  用户在不同的月份都有订单记录 ,如果查询所有订单 就需要去多个库查询,性能很差
     *
     *
     *
     *  charging_bill_success_区域_华北
     *  charging_bill_success_区域_东北
     *
     *   用户的app 我的订单 订单列表
     *     用户在不同的区域都有订单记录 情况比较少 ,如果查询所有订单 不需要去多个库查
     *
     *  按照区域存储
     *
     *  charging_bill_success_bak 超过半年 放到另外一个数据库 冷数据  机器配置低
     *
     *  保留半年
     *  48W * 180   =  8640W  mysql 单表存 千万级以内  亿级 能存, 容量限制 todo
     *
     * 非关系型
     * 分布式 多个节点 同时提供 数据的 存 储
     *
     * Es 分布式搜索引擎
     * 能存
     *
     *
     * @param chargingProcessParam
     */
    private void saveChargingProcess(ChargingProcessParam chargingProcessParam) {
        log.debug("保存充电进度数据:{}",chargingProcessParam);
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        //时间
        // 时序数据 时间 顺序 数据
        // 日志 程序日志
        // 16:29:01 同步1次  张三
        // 16:29:02 同步1次  张三
        // 16:29:03 同步5次  张三 李四 王五 赵六 钱七

        //11:10:21 股票 +1
        //11:10:22 股票 +1
        //11:10:23 股票 +3

        //天气预报
        //11:10:21 云厚度 +1
        //11:10:22 云厚度 +1
        //11:10:23 云厚度 +2

        //时间
        chargingProcessPO.setId(System.currentTimeMillis());
        //数据
        BeanUtils.copyProperties(chargingProcessParam,chargingProcessPO);
        //保存到es
        mqttContext.getChargingProcessRepository().save(chargingProcessPO);
        log.debug("保存充电进度数据成功:{}",chargingProcessParam);
    }

    private void handleChargingResult(MqttMessage message) {
        ChargingResultDto chargingResultDto = null;
        try {
            //把设备发送的消息JSON转换java对象ChargingResultDto
            chargingResultDto =
                    JsonUtils.fromJson(message.toString(), ChargingResultDto.class);
            log.debug("消息转换为java对象:{}", chargingResultDto);
        } catch (Exception e) {
            log.error("设备发送的开始充电结果消息格式{}有问题,请检查!", message);
        }
        //判断java对象不能为空 chargingResultDto
        if (chargingResultDto != null) {
            //获取设备充电结果
            String result = chargingResultDto.getResult();
            //判断设备充电结果
            if (Constant.RESULT_START_CHARGING_SUCCESS.equals(result)) {
                //设备开始充电成功
                log.debug("充电桩开始充电成功");
                ChargingBillSuccessPO chargingBillSuccessPO = new ChargingBillSuccessPO();
                chargingBillSuccessPO.setBillId(chargingResultDto.getOrderNo());
                chargingBillSuccessPO.setBillStatus(OrderStatusConstant.ORDER_STATUS_PROGRESS);
                log.debug("保存成功订单记录:{}", chargingBillSuccessPO);
                //如果设备重复发送消息
                //会导致 成功订单表 charging_bill_success 会重复的保存订单数据
                //如何解决
                //1 给订单表 订单编号 创建一个唯一索引,唯一索引的意思 这个表里不能有重复的订单编号
                // 加完唯一索引后,如果有重复的记录要保存 数据库会抛出异常
                // Duplicate entry '10000_58_1721720859912' for key 'charging_bill_success_bill_id_unique_key'
                // 从而避免订单成功表有重复的订单记录
                // 异常可以处理,可以不处理 通常接入监控系统  监控系统
                // 如果固定的时间之内 重复发送消息,设备出问题了,通知调用方 设备维修人员
                //2 先查询订单成功表是否有数据,如果没有 插入,如果有 跳过不保存,
                // 代码怎么写? 原子性的问题  todo
                //3 记录状态 如果处理过了,不再重复处理  redis的布隆过滤器 自己了解 todo

                try {
                    Integer row = mqttContext.getOrderSuccessMapper().insert(chargingBillSuccessPO);
                    log.debug("保存成功订单记录:{},完成,影响行数:{}", chargingBillSuccessPO, row);
                } catch (Exception e) {
                    //接入监控系统
                    log.error("保存成功订单数据失败", e);
                }
                //通过订单编号查询订单数据
                //T1 10:25:01:01
                //Integer count = getOrderByOrderNo(chargingResultDto.getOrderNo());
                // count = 0
                //T2 10:25:01:02
                //Integer count = getOrderByOrderNo(chargingResultDto.getOrderNo());
                // count = 0
                //T1 10:25:01:02 判断数据行数 count == 0 true
                //T2 10:25:01:03 判断数据行数 count == 0 true
                //T1 和 T2 判断 count == 0 都为真 都执行了插入动作
                //if (count == 0) {
                //行数等于0 ,表示数据库没有 ,保存
                // T1 insert()
                // T2 insert()
                // 重复的记录
                //    insert();
                //}else {
                //不等于0,处理过,无需任何处理
                //    log.debug("订单编号:{},数据已经存在,无需处理",chargingResultDto.getOrderNo());
                //}
                //加锁 保证线程来排队 T1 先执行,T2 再执行 ,性能下降


            } else {
                //不是成功,都是失败
                log.debug("充电桩开始充电失败");
                ChargingBillFailPO chargingBillFailPO = new ChargingBillFailPO();
                chargingBillFailPO.setBillId(chargingResultDto.getOrderNo());
                chargingBillFailPO.setFailDesc("充电桩故障-无法正常充电");
                log.debug("保存失败订单记录:{}", chargingBillFailPO);
                Integer row = mqttContext.getOrderFailMapper().insert(chargingBillFailPO);
                log.debug("保存失败订单记录:{},完成,影响行数:{}", chargingBillFailPO, row);
            }
        }
    }


    /**
     * 消息处理完成
     *
     * @param token the delivery token associated with the message.
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("MqttConsumer#deliveryComplete方法被调用了,表示消息处理完成");
    }

    /**
     * 调用计价服务计算价格
     *
     * @param chargingProcessParam
     * @return
     */
    private ChargingProcessVO getOrderCost(ChargingProcessParam chargingProcessParam) {
        JsonResult<ChargingProcessVO> chargingProcessVOJsonResult =
                mqttContext.getCostClient().calculateCost(chargingProcessParam);
        if (chargingProcessVOJsonResult != null) {
            ChargingProcessVO data = chargingProcessVOJsonResult.getData();
            if (data != null) {
                return data;
            }
        }
        return null;
    }
}